package com.zhao.sparkstreaming.sgg.code

import java.sql.Connection

import com.zhao.sparkstreaming.sgg.code.handler.{BlackListHandler, DateAreaCityAdCountHandler, LastHourAdCountHandler}
import com.zhao.sparkstreaming.sgg.utils.{JdbcUtil, MyKafkaUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}


/**
 * Description: <br/>
 * Copyright (c) ，2020 ， 赵 <br/>
 * This program is protected by copyright laws. <br/>
 * Date： 2020/12/4 16:35
 *
 * @author 柒柒
 * @version : 1.0
 */

object RealTimeApp {
  def main(args: Array[String]): Unit = {

    //1.创建SparkConf
    val conf: SparkConf = new SparkConf()
      .setAppName(this.getClass.getSimpleName)
      .setMaster("local[*]")

    //2.创建StreamingContext
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))

    //3.读取数据
    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream("test", ssc)

    //4.将从kafka读出的数据转换为样例类对象
    val adsLogDStream: DStream[Ads_log] = kafkaDStream.map(record => {
      val value: String = record.value()
      val arr: Array[String] = value.split(" ")
      Ads_log(arr(0).toLong, arr(1), arr(2), arr(3), arr(4))
    })

//    //5.需求一:根据Mysql中的黑名单过滤当前数据集
//    val filterAdsLogDStream: DStream[Ads_log] = BlackListHandler.filterByBlackList(adsLogDStream)
//
//    //6.需求二:将满足要求的用户写入黑名单
//    BlackListHandler.addBlackList(filterAdsLogDStream)
//
//    //测试打印
//    filterAdsLogDStream.cache()
//    filterAdsLogDStream.count().print()

    //5.根据Mysql中的黑名单表进行数据过滤
    val filterAdsLogDStream: DStream[Ads_log] = adsLogDStream.filter(adsLog => {
      //查询Mysql,查看当前用户是否存在
      val connection: Connection = JdbcUtil.getConnection
      val bool: Boolean = JdbcUtil.isExist(connection, "select * from black_list where userid=?", Array(adsLog.userid))
      connection.close()
      !bool
    })

    filterAdsLogDStream.cache()

    //6.对没有被加入黑名单的用户统计当前批次单日各个用户对各个广告点击的总次数
    //并更新值Mysql
    //之后查询更新之后的数据,判断是否超过100次
    //如果超过则将用户加入黑名单
    BlackListHandler.addBlackList(filterAdsLogDStream)

    //7.统计每天各大区各个城市广告点击总数并保存至mysql中
    DateAreaCityAdCountHandler.saveDateAreaCityAdCountToMysql(filterAdsLogDStream)

    //8.统计最近一小时(2分钟)广告点击总数
    val adToHmCountListDStream: DStream[(String, List[(String, Long)])] = LastHourAdCountHandler.getAdHourMintToCount(filterAdsLogDStream)

    //9.打印
    adToHmCountListDStream.print()
    //10.启动任务
    ssc.start()
    ssc.awaitTermination()
  }
}















